package yunjiao.javatutorials.apache.kafka.component;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import yunjiao.javatutorials.apache.kafka.model.InventoryUpdateEvent;
import yunjiao.javatutorials.apache.kafka.model.NotificationEvent;
import yunjiao.javatutorials.apache.kafka.model.OrderCreatedEvent;

import java.util.Map;

/**
 * 订单事件消费者
 *
 * @author yangyunjiao
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class OrderEventConsumer {
    private final KafkaMessageProducer kafkaMessageProducer;

    @KafkaListener(
            topics = "${app.kafka.topics.order-created}",
            groupId = "${app.kafka.consumer-groups.inventory-service}",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void consumeOrderCreatedEvent(OrderCreatedEvent event,
                                         Acknowledgment ack) {
        try {
            log.info(">>> Received order created event: orderId={}, userId={}",
                    event.getOrderId(), event.getUserId());

            // 处理库存扣减
            processInventoryUpdate(event);

            // 发送确认通知
            sendOrderConfirmation(event);

            // 手动确认消息
            ack.acknowledge();
            log.info(">>> Order created event processed successfully: orderId={}",
                    event.getOrderId());

        } catch (Exception e) {
            log.error("Error processing order created event: orderId={}",
                    event.getOrderId(), e);
            // 根据业务需求决定是否重试或进入死信队列
        }
    }

    private void processInventoryUpdate(OrderCreatedEvent event) {
        event.getItems().forEach(item -> {
            InventoryUpdateEvent inventoryEvent = new InventoryUpdateEvent();
            inventoryEvent.setProductId(item.getProductId());
            inventoryEvent.setQuantity(item.getQuantity());
            inventoryEvent.setOperation(InventoryUpdateEvent.OperationType.DECREASE);
            inventoryEvent.setOrderId(event.getOrderId());

            // 发送库存更新事件
            kafkaMessageProducer.sendInventoryUpdateEvent(inventoryEvent);
        });
    }

    private void sendOrderConfirmation(OrderCreatedEvent event) {
        NotificationEvent notification = new NotificationEvent();
        notification.setUserId(event.getUserId());
        notification.setType(NotificationEvent.NotificationType.ORDER_CONFIRMED);
        notification.setTitle("订单确认");
        notification.setContent("您的订单已成功创建，订单号: " + event.getOrderId());
        notification.setAdditionalData(Map.of("orderId", event.getOrderId(),
                "amount", event.getAmount()));

        // 发送通知事件
        kafkaMessageProducer.sendNotificationEvent(notification);
    }
}
